package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import cn.gwm.utils.HBaseDataSource;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @author : GW00280745
 * @date : 2023/4/26 13:05
 * @description :
 * 实时更新每1%soc的时间到phoenix中
 *
 */
public class OneSocKeyFunction extends KeyedProcessFunction<Object, JSONObject, JSONObject> {
    @Override
    public void processElement(JSONObject line, KeyedProcessFunction<Object, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
        String vinVal = line.getString(ChargePredictFields.ins().srcFieldVin);
        Long tidVal = line.getLong(ChargePredictFields.ins().srcFieldTid);
        String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
        Float bmsSoc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
        //先判断状态是否有数据
        String socKey = vinVal.concat(startChargeTime);
        boolean contains = oneSocState.contains(socKey);
        //没有，则添加到state中
        String concat = tidVal.toString().concat(ChargePredictFields.ins().char3).concat(bmsSoc.toString());
        if (!contains){
            oneSocState.put(socKey,concat);
        }
        //有，则进行判断 soc差值是否为整数，如果是整数则更新phoenix
        String s = oneSocState.get(socKey);
        String[] split = s.split(ChargePredictFields.ins().char3);
        String tidState = split[0];
        String socState = split[1];
        boolean b = deltaSoc(bmsSoc, socState);
        if (!b){
            return;
        }
        oneSocState.put(socKey,concat);
        //更新phoenix
        float e=oneSocTime(tidVal,tidState,bmsSoc, socState);
        String id = vinVal.concat(kangStr).concat(StringUtils.reverse(tidVal.toString()));
    }

    private String kangStr= "-";
    private int num1=1;
    private int num2=2;
    private int num3=3;
    private int num4=4;
    private int num10=10;
    private String str100="100";
    private String str100of0="100.0";

    private String socConvert(String soc){
        int length = soc.length();
        if (length==num1){
            return  "00".concat(soc).concat(".0");
        }else if (length == num2){
            return  "0".concat(soc).concat(".0");
        }else if (length == num3){
            if (StringUtils.equalsIgnoreCase(soc,str100) || StringUtils.equalsIgnoreCase(soc,str100of0)){
                return "100.0";
            }else {
                return "00".concat(soc);
            }
        }else if (length == num4){
            return  "0".concat(soc);
        }
        return null;
    }
    private String tempConvert(String tempMin){
        float v = Float.parseFloat(tempMin);
        float abs = Math.abs(v);

        if (v<0){
            if (abs<num10){
                return "-0".concat(String.valueOf((int)abs));
            }else {
                return tempMin;
            }
        }else if (v>0){
            if (abs<num10){
                return "a0".concat(String.valueOf((int)abs));
            }else {
                return "a".concat(String.valueOf((int)abs));
            }
        }else {
            return "a00";
        }
    }
    private String currConvert(String curr){
        float v = Float.parseFloat(curr);
        float abs = Math.abs(v);
        if (abs<num10){
            return "-0".concat(String.valueOf(abs));
        }
        return curr;
    }
    private float oneSocTime(Long tidVal, String tidState, Float bmsSoc, String socState) {
        long tidStateL = Long.parseLong(tidState);
        float socStateF = Float.parseFloat(socState);
        BigDecimal bigDecimal = new BigDecimal((tidVal - tidStateL) / (bmsSoc - socStateF));
        float v = bigDecimal.abs().divide(new BigDecimal(60000), 1, BigDecimal.ROUND_HALF_UP).floatValue();
        return v;
    }

    private boolean deltaSoc(Float bmsSoc, String socState) {
        float v = Float.parseFloat(socState);
        float delta = bmsSoc - v;
        //判断 soc差值是否为整数 找大于0.9的第一个值
        boolean b = delta >= 0.9 ;
        return b;
    }

    private MapState<String, String> oneSocState;
    @Override
    public void open(Configuration parameters) throws Exception {
        oneSocState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("oneSocState", String.class, String.class));
    }

}
